package poolserver

import (
	"fmt"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	config "csudata.com/zqpool/src/config"
	"go.uber.org/zap"
)

const (
	BE_BAD_NO_FREE int8 = -1 //表示此portal已经坏了，但可能还有未被释放
	BE_BAD_FREEED  int8 = -2 //表示此portal已经坏了，但所有连接都被释放了
	BE_REMOVING    int8 = -9 //表示此portal正在被移除
	BE_OFFLINE     int8 = 0  //空闲未被上线使用
	BE_ONLINE      int8 = 1  //上线使用中
)

const (
	DB_READONLY  int8 = 1 //只读节点
	DB_READWRITE int8 = 2 //读写节点
)

type BePool struct {
	State    int8                //记录Portal的状态，可以为BE_BAD_NO_FREE, BE_BAD_FREEED, BE_REMOVING, BE_NO_USED
	RwState  int8                //记录Portal的状态连接的是可以读写的主库，还是只读的备库, BE_READONLY, BE_READWRITE
	BeConns  []*PgBackendConn    //记录到后端的所有连接
	FreeChan chan *PgBackendConn //未被使用的连接放到此channel中
	Lock     sync.Mutex
}

type BePoolStatus struct {
	Portal  string
	State   string
	RwState string
}

func (bep *BePool) Init() {
	bep.BeConns = make([]*PgBackendConn, 0, 100)
	bep.FreeChan = make(chan *PgBackendConn, 5000)
}

func (bep *BePool) GetState() int8 {
	var beState int8
	bep.Lock.Lock()
	defer bep.Lock.Unlock()
	beState = bep.State
	return beState
}

func (bep *BePool) SetState(state int8) {
	bep.Lock.Lock()
	defer bep.Lock.Unlock()
	bep.State = state
}

func PoolGetBeDbStatusList(poolName string) ([]*BePoolStatus, error) {
	PoolsLock.RLock()
	pool, ok := g_allpools[poolName]
	PoolsLock.RUnlock()
	if !ok {
		return nil, fmt.Errorf("pool(%s) not exists", poolName)
	}
	pool.Lock.RLock()
	defer pool.Lock.RUnlock()
	BePoolStatusList := make([]*BePoolStatus, 0, len(pool.BePools))
	for portal, bePool := range pool.BePools {
		BePoolStatus := new(BePoolStatus)
		BePoolStatus.Portal = portal

		switch {
		case bePool.State == BE_BAD_NO_FREE:
			BePoolStatus.State = "BAD(some connect not free)"
		case bePool.State == BE_BAD_FREEED:
			BePoolStatus.State = "BAD(all connect freed)"
		case bePool.State == BE_REMOVING:
			BePoolStatus.State = "Removing"
		case bePool.State == BE_OFFLINE:
			BePoolStatus.State = "Offline"
		case bePool.State == BE_ONLINE:
			BePoolStatus.State = "Online"
		default:
			BePoolStatus.State = "Unknown"
		}
		if bePool.RwState == DB_READONLY {
			BePoolStatus.RwState = "Standby"
		} else {
			BePoolStatus.RwState = "Primary"
		}
		BePoolStatusList = append(BePoolStatusList, BePoolStatus)
	}
	return BePoolStatusList, nil
}

// PoolAddHost 某个池增加一个后端的数据库主机
func PoolAddHost(poolName string, portal string) error {
	var err error
	PoolsLock.RLock()
	pool, ok := g_allpools[poolName]
	PoolsLock.RUnlock()
	if !ok {
		return fmt.Errorf("pool(%s) not exists", poolName)
	}
	bePool := new(BePool)
	bePool.Init()
	bePool.State = BE_ONLINE
	err = BuildBePoolConn(pool, portal, bePool)
	if err != nil {
		return err
	}
	pool.Lock.Lock()
	_, ok = pool.BePools[portal]
	if ok {
		pool.Lock.Unlock()
		for _, beConn := range bePool.BeConns {
			beConn.Conn.Close()
		}
		close(bePool.FreeChan)
		return fmt.Errorf("portal(%s) already in pool(%s)", portal, poolName)
	}
	pool.BePools[portal] = bePool
	pool.Lock.Unlock()
	return nil
}

func (pool *Pool) RemoveHost(portal string) error {
	var err error

	pool.Lock.Lock()
	defer pool.Lock.Unlock()
	bePool, ok := pool.BePools[portal]
	if !ok {
		return fmt.Errorf("portal(%s) not exists", portal)
	}
	/* 当移除一个节点时，可以把后端的空闲连接给关闭掉，对于活动连接，等其释放后，也关闭掉*/
	bePool.Lock.Lock()
	bePool.State = BE_REMOVING
	bePool.Lock.Unlock()
	return err
}

func BuildBePoolConn(pool *Pool, portal string, bePool *BePool) error {
	//var bePool *BePool = pool.BePools[portal]
	var i int
	var rw_state int8
	var serverVersion string
	var cnt int

	// if bePool == nil {
	// 	return fmt.Errorf("portal(%s) not exists", portal)
	// }

	conn, err, _, _ := connectBackend(portal, pool.BeUser, pool.BePasswd, pool.BeDbName)
	if err != nil {
		zap.S().Errorf("Could not connect %s: %s", portal, err.Error())
		return err
	}

	//_, timeline_id, lsn := getTimelineAndLsn(conn)
	//zap.S().Infof("timeline_id=%d, lsn=%d", timeline_id, lsn)

	err, serverVersion = getDbServerVersion(conn)
	if err != nil {
		zap.S().Errorf("Could not get database version from %s: %s", portal, err.Error())
		return err
	}
	zap.S().Infof("Pool portal(%s) ServerVersion=%s", portal, serverVersion)
	pool.DbVersion = serverVersion

	err, rw_state = getDbReadWriteState(conn)
	if err != nil {
		zap.S().Errorf("Could not get database is primary or standby from %s: %s", portal, err.Error())
		return err
	}
	conn.Close()
	if rw_state == 2 {
		cnt = int(pool.BeRwConnCount)
	} else {
		cnt = int(pool.BeRdConnCount)
	}

	bePool.BeConns = make([]*PgBackendConn, 0, 100)
	bePool.RwState = rw_state

	for i = 0; i < cnt; i++ {
		conn, err, pid, bekey := connectBackend(portal, pool.BeUser, pool.BePasswd, pool.BeDbName)
		if err != nil {
			zap.S().Errorf("Could not connect %s: %s", portal, err.Error())
			for i = i - 1; i >= 0; i-- {
				bePool.BeConns[i].Conn.Close()
			}
			return err
		}
		pBackend := new(PgBackendConn)
		pBackend.Init()
		pool.BeMaxId++
		pBackend.Id = pool.BeMaxId
		pBackend.Pid = pid
		pBackend.BeKey = bekey
		pBackend.Conn = conn
		pBackend.ConnPortal = portal
		pBackend.ConnUnixTime = time.Now().Unix()
		bePool.BeConns = append(bePool.BeConns, pBackend)
		//初始时，把所有后端连接都放到空闲队列中
		bePool.FreeChan <- pBackend
	}
	return nil
}

func (pool *Pool) ModifyFeInfo(FeUser, FePasswd, FeDbName, poolName string) (int, string) {
	pool.Lock.Lock()
	// check the new information is changed or not
	if pool.FeUser == FeUser && pool.FePasswd == FePasswd && pool.FeDbName == FeDbName {
		pool.Lock.Unlock()
		return -1, "The new information is the same as the old,please check."
	}

	// update pool info
	if FeUser != "" {
		pool.FeUser = FeUser
	}
	if FePasswd != "" {
		pool.FePasswd = FePasswd
	}
	if FeDbName != "" {
		pool.FeDbName = FeDbName
	}
	// Add new pool and remove old pool
	new_poolName := pool.FeUser + "." + pool.FeDbName
	if new_poolName != poolName {
		g_allpools[new_poolName] = pool
		delete(g_allpools, poolName)
	}
	pool.Lock.Unlock()
	return 0, "Success modify the pool information."
}

func (pool *Pool) ModifyBeInfo(BeUser, BePasswd string) (int, string) {
	var is_right bool

	pool.Lock.Lock()

	// check the new information is changed or not
	if pool.BeUser == BeUser && pool.BePasswd == BePasswd {
		pool.Lock.Unlock()
		return -1, "The new information is the same as the old,please check."
	}

	// test connect to check the new information is ok
	for portal := range pool.BePools {
		conn, err, _, _ := connectBackend(portal, BeUser, BePasswd, pool.BeDbName)
		if err != nil {
			err_msg := "Could not connect" + portal + err.Error() + ",Please check."
			zap.S().Errorf(err_msg)
			pool.Lock.Unlock()
			return -1, err_msg
		}
		err_conn, _ := getDbServerVersion(conn)
		if err_conn != nil {
			err_msg := fmt.Sprintf("Could not connect %s, %s, Please check the user and password.", portal, err_conn.Error())
			zap.S().Errorf(err_msg)
			pool.Lock.Unlock()
			return -1, err_msg
		}
		conn.Close()
		is_right = true
		break

	}
	if is_right {
		// update pool info
		if BeUser != "" {
			pool.BeUser = BeUser
		}
		if BePasswd != "" {
			pool.BePasswd = BePasswd
		}
	}
	pool.Lock.Unlock()
	return 0, "Success modify the pool information."
}

type poolInfo struct {
	ID            string
	FeMaxConns    int32
	FeUser        string
	FePasswd      string
	FeDbName      string
	BeRwConnCount int32
	BeRdConnCount int32
	BeUser        string
	BePasswd      string
	BeDbName      string
	DbVersion     string /*后端数据库的版本*/
	MgrAddr       string
	ListenAddr    string
	BePortals     []string
}

func PoolToPoolInfo(pool *Pool) *poolInfo {
	retPool := new(poolInfo)
	// pool.BePools = nil
	retPool.ID = pool.ID
	retPool.FeUser = pool.FeUser
	retPool.FePasswd = pool.FePasswd
	retPool.FeDbName = pool.FeDbName
	retPool.FeMaxConns = atomic.LoadInt32(&pool.FeMaxConns)
	retPool.BeUser = pool.BeUser
	retPool.BePasswd = pool.BePasswd
	retPool.BeDbName = pool.BeDbName
	retPool.DbVersion = pool.DbVersion
	retPool.BeRwConnCount = pool.BeRwConnCount
	retPool.BeRdConnCount = pool.BeRdConnCount

	retPool.ListenAddr = config.Get("listen_addr")
	retPool.MgrAddr = config.Get("mgr_addr")

	// get the BeConnProtals
	for portal := range pool.BePools {
		retPool.BePortals = append(retPool.BePortals, portal)
	}

	return retPool
}

func GetPoolList(poolName string) (int, []*poolInfo) {

	poolMap := make(map[string]*poolInfo)
	// filter: poolName or ""

	if len(poolName) != 0 {
		pool := GetPoolByName(poolName)
		if pool == nil {
			return -1, nil
		}
		retPool := PoolToPoolInfo(pool)
		poolMap[retPool.ID] = retPool
	} else {
		for _, pool := range g_allpools {
			retPool := PoolToPoolInfo(pool)
			poolMap[retPool.ID] = retPool
		}
	}

	poolList := make([]*poolInfo, 0, len(poolMap))
	for _, poolInfo := range poolMap {
		poolList = append(poolList, poolInfo)
	}

	return 0, poolList
}

func RemovePool(poolName string) error {

	pool := GetPoolByName(poolName)
	if pool == nil {
		return fmt.Errorf("the pool(poolName=%s) is not exists", poolName)
	}

	// run go fuc, check the BeConn to empty
	go func() {
		for portal := range pool.BePools {
			pool.RemoveHost(portal)
		}

		for {
			time.Sleep(2 * time.Second)
			bePool_num := len(pool.BePools)
			if bePool_num == 0 {
				PoolsLock.Lock()
				delete(g_allpools, poolName)
				PoolsLock.Unlock()
				break
			}
		}
	}()

	return nil
}

func ModifyPoolFeMaxConns(pool *Pool, fe_max_conns string) (int, string) {
	var err error
	var iVa int
	var FeMaxConns int32

	// check the fe_max_conns
	if fe_max_conns != "" {
		iVa, err = strconv.Atoi(fe_max_conns)
		if err != nil {
			return -1, fmt.Sprintf("invalid parameter: %s, must be a number!", fe_max_conns)
		}
		FeMaxConns = int32(iVa)
		// fe_max_conns can reduce
		atomic.StoreInt32(&pool.FeMaxConns, FeMaxConns)
	}
	return 0, ""
}

func RollbackBeAddConns(bePoolList []*BePool, old_len int, new_conns int) {
	var i int
	for _, bePool := range bePoolList {
		for i = new_conns - 1; i >= old_len; i-- {
			bePool.BeConns[i].Conn.Close()
		}
		bePool.BeConns = bePool.BeConns[:old_len]
	}
}

func ModifyPoolBeConns(pool *Pool, be_conns string, rw_state int8) (int, string) {
	var err error
	var iVa int
	var BeConns int32
	var OldBeConns int32
	var oldBeLen int
	var i int

	if be_conns == "" {
		return 0, ""
	}

	iVa, err = strconv.Atoi(be_conns)
	if err != nil {
		return -1, fmt.Sprintf("invalid parameter: %s, must be a number!", be_conns)
	}
	BeConns = int32(iVa)
	if rw_state == DB_READWRITE {
		OldBeConns = atomic.LoadInt32(&pool.BeRwConnCount)
	} else {
		OldBeConns = atomic.LoadInt32(&pool.BeRdConnCount)
	}
	if BeConns <= OldBeConns {
		return -1, fmt.Sprintf("Can only increase be_rw_conns, new value(%d) must more than old value=%d!", BeConns, OldBeConns)
	}

	// fe_max_conns can reduce
	//atomic.StoreInt32(&pool.FeMaxConns, FeMaxConns)

	pool.Lock.Lock()
	// 把处理过的bePool放到bePoolList，方便出错后回滚
	bePoolList := make([]*BePool, 0, 3)
	for portal, bePool := range pool.BePools {
		bePool.Lock.Lock()
		if bePool.RwState != rw_state || bePool.State != BE_ONLINE {
			bePool.Lock.Unlock()
			continue
		}

		//oldBeLen = len(bePool.BeConns)
		bePoolList = append(bePoolList, bePool)
		for i = int(OldBeConns); i < int(BeConns); i++ {
			conn, err, pid, bekey := connectBackend(portal, pool.BeUser, pool.BePasswd, pool.BeDbName)
			if err != nil {
				zap.S().Errorf("Could not connect %s: %s", portal, err.Error())
				for i = i - 1; i >= oldBeLen; i-- {
					bePool.BeConns[i].Conn.Close()
				}
				bePool.BeConns = bePool.BeConns[:oldBeLen]
				if len(bePoolList) > 1 {
					bePoolList = bePoolList[0 : len(bePoolList)-1]
					RollbackBeAddConns(bePoolList, oldBeLen, int(BeConns))
				}
				bePool.Lock.Unlock()
				pool.Lock.Unlock()
				return -1, err.Error()
			}
			pBackend := new(PgBackendConn)
			pBackend.Init()
			pool.BeMaxId++
			pBackend.Id = pool.BeMaxId
			pBackend.Pid = pid
			pBackend.BeKey = bekey
			pBackend.Conn = conn
			pBackend.ConnPortal = portal
			pBackend.ConnUnixTime = time.Now().Unix()
			bePool.BeConns = append(bePool.BeConns, pBackend)
		}
		bePool.Lock.Unlock()
	}

	//把所有新增的后端连接都放到空闲队列中
	for _, bePool := range pool.BePools {
		bePool.Lock.Lock()
		if bePool.RwState != rw_state || bePool.State != BE_ONLINE {
			bePool.Lock.Unlock()
			continue
		}

		for i = int(OldBeConns); i < int(BeConns); i++ {
			bePool.FreeChan <- bePool.BeConns[i]
		}
		bePool.Lock.Unlock()
	}

	// 完全成功后，把连接的值设置上去
	if rw_state == DB_READWRITE {
		atomic.StoreInt32(&pool.BeRwConnCount, BeConns)
	} else {
		atomic.StoreInt32(&pool.BeRdConnCount, BeConns)
	}
	pool.Lock.Unlock()
	return 0, ""
}

func ModifyPoolConns(poolName, fe_max_conns, be_rw_conns, be_rd_conns string) (int, string) {
	//var err error

	pool, ok := g_allpools[poolName]
	if !ok {
		return -1, fmt.Sprintf("The pool(poolName=%s) is not exists.", poolName)
	}

	errCode, errMsg := ModifyPoolFeMaxConns(pool, fe_max_conns)
	if errCode != 0 {
		return errCode, errMsg
	}

	errCode, errMsg = ModifyPoolBeConns(pool, be_rw_conns, DB_READWRITE)
	if errCode != 0 {
		return errCode, errMsg
	}

	errCode, errMsg = ModifyPoolBeConns(pool, be_rd_conns, DB_READONLY)
	if errCode != 0 {
		return errCode, errMsg
	}

	return 0, "Success modify the connects information."
}
